package com.hskn.hss.utils;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaProducer<K, V> {
    public static final String CONFIG_PRODUCER_FILE_NAME = "kafka.producer.properties";

    private Producer<K, V> producer;

    KafkaProducer(String path) {
        Properties props = new Properties();
        try {
            InputStream in = new BufferedInputStream(new FileInputStream(path));
            props.load(in);
        } catch (IOException e) {
            e.printStackTrace();
            return;
        }
        producer = new org.apache.kafka.clients.producer.KafkaProducer(props);
    }

    public KafkaProducer() {
        Properties props = new Properties();
        try {
            props = Tools.loadFromClasspath(CONFIG_PRODUCER_FILE_NAME, KafkaProducer.class);
        } catch (IOException e) {
            e.printStackTrace();
            return;
        }
        producer = new org.apache.kafka.clients.producer.KafkaProducer(props);
    }

    public KafkaProducer(Map<String, Object> configs) {
        producer = new org.apache.kafka.clients.producer.KafkaProducer(configs);
    }

    /**
     * 生产消息
     *
     * @param topic     topic对象
     * @param partition partition
     * @param key       消息key
     * @param data      消息数据
     */
    public void produce(String topic, Integer partition, K key, V data) {
        produce(topic, partition, key, data, null, (Callback) null);
    }

    /**
     * 生产消息
     *
     * @param topic     topic对象
     * @param partition partition
     * @param key       消息key
     * @param data      消息数据
     * @param timestamp timestamp
     */
    public void produce(String topic, Integer partition, K key, V data, Long timestamp) {
        produce(topic, partition, key, data, timestamp, (Callback) null);
    }

    /**
     * 生产消息
     *
     * @param topic     topic对象
     * @param partition partition
     * @param key       消息key
     * @param data      消息数据
     * @param callback  callback
     */
    public void produce(String topic, Integer partition, K key, V data, Callback callback) {
        produce(topic, partition, key, data, null, callback);
    }

    public void produce(String topic, V data) {
        produce(topic, null, null, data, null, (Callback) null);
    }

    /**
     * 生产消息
     *
     * @param topic     topic对象
     * @param partition partition
     * @param key       消息key
     * @param data      消息数据
     * @param timestamp timestamp
     * @param callback  callback
     */
    public void produce(String topic, Integer partition, K key, V data, Long timestamp, Callback callback) {
        ProducerRecord<K, V> kafkaRecord =
                timestamp == null ? new ProducerRecord<K, V>(topic, partition, key, data)
                        : new ProducerRecord<K, V>(topic, partition, timestamp, key, data);
        produce(kafkaRecord, callback);
    }

    public void produce(ProducerRecord<K, V> kafkaRecord) {
        produce(kafkaRecord, (Callback) null);
    }

    public void produce(ProducerRecord<K, V> kafkaRecord, Callback callback) {
        producer.send(kafkaRecord, callback);
    }

    public void close() {
        producer.close();
    }
}
